import requests
import json
import time
import pymysql
from datetime import datetime
import pandas as pd
from sklearn.impute import KNNImputer

def weather_data_pipeline(city_list, api_key, db_config):
    """
    天气数据采集、清洗和存储的完整流程

    参数:
        city_list (dict): 城市名称和对应ID的字典
        api_key (str): 天气API的密钥
        db_config (dict): 数据库连接配置
    """
    # 创建数据库表
    create_weatherdata_table(db_config)
    create_cleaned_weatherdata_table(db_config)

    # 清空表数据
    truncate_table("weatherdata", db_config)
    truncate_table("cleaned_weatherdata", db_config)

    # 数据采集部分
    for city_name, city_id in city_list.items():
        current_weather_data = crawl_current_weather(city_id, api_key)
        save_current_weather_to_database(city_name, current_weather_data, db_config)

    # 数据清洗部分
    clean_weather_data(db_config)

def crawl_current_weather(city_id, api_key):
    """爬取指定城市的当前天气数据"""
    base_url = "https://devapi.qweather.com/v7/weather/now"
    full_url = f"{base_url}?key={api_key}&location={city_id}"

    try:
        response = requests.get(full_url)
        data = response.json()

        if data["code"] == "200":
            now_data = data["now"]
            weather_info = {
                "obs_time": now_data.get("obsTime", ""),
                "temperature": float(now_data.get("temp", 0)),
                "feels_like": float(now_data.get("feelsLike", 0)),
                "text": now_data.get("text", ""),
                "wind_360": float(now_data.get("wind360", 0)),
                "wind_dir": now_data.get("windDir", ""),
                "wind_scale": now_data.get("windScale", ""),
                "wind_speed": float(now_data.get("windSpeed", 0)),
                "humidity": float(now_data.get("humidity", 0)),
                "precipitation_1h": float(now_data.get("precip", 0)),
                "pressure": float(now_data.get("pressure", 0)),
                "visibility": float(now_data.get("vis", 0))
            }
            return weather_info
        else:
            print(f"Error: {data['code']}, {data['message']}")
            return None

    except Exception as e:
        print(f"Error: {e}")
        return None

def save_current_weather_to_database(city_name, current_weather_data, db_config):
    """将当前天气数据保存到数据库"""
    try:
        current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')

        if current_weather_data:
            obs_time = current_weather_data.get("obs_time")
            if obs_time:
                dt = datetime.fromisoformat(obs_time.replace('Z', '+00:00'))
                obs_time = dt.strftime('%Y-%m-%d %H:%M:%S')

            data = (
                city_name,
                obs_time,
                current_weather_data.get("temperature"),
                current_weather_data.get("feels_like"),
                current_weather_data.get("text"),
                current_weather_data.get("wind_360"),
                current_weather_data.get("wind_dir"),
                current_weather_data.get("wind_scale"),
                current_weather_data.get("wind_speed"),
                current_weather_data.get("humidity"),
                current_weather_data.get("precipitation_1h"),
                current_weather_data.get("pressure"),
                current_weather_data.get("visibility")
            )
        else:
            data = (
                city_name,
                current_time,
                None, None, None,
                None, None, None,
                None, None, None,
                None, None
            )

        insert_sql = """
        INSERT INTO weatherdata (
            城市, 时间, 温度, 体感温度, 天气情况,
            风向角度, 风向, 风力等级, 风速, 湿度, 降水量, 大气压强, 能见度
        ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
        """

        try:
            connection = pymysql.connect(**db_config)
            with connection.cursor() as cursor:
                cursor.execute(insert_sql, data)
            connection.commit()
            if current_weather_data:
                print(f"当前天气数据保存成功: {city_name}")
            else:
                print(f"当前天气数据获取失败，但已保存城市和时间: {city_name}")
        except Exception as e:
            print(f"保存 {city_name} 天气数据到数据库时出错: {e}")
            connection.rollback()
        finally:
            connection.close()

    except (TypeError, ValueError) as err:
        print(f"数据处理错误: {err}")

def clean_weather_data(db_config):
    """清洗天气数据"""
    try:
        connection = pymysql.connect(**db_config)

        query_sql = "SELECT * FROM weatherdata"
        df = pd.read_sql(query_sql, connection)

        print("原始数据:")
        print(df.head())
        print("\n数据基本信息:")
        df.info()
        print("\n缺失值情况:")
        print(df.isnull().sum())

        # 处理缺失值
        numeric_columns = ['温度', '体感温度', '风向角度', '风速', '湿度', '降水量', '大气压强', '能见度']
        if any(df[numeric_columns].isnull().any()):
            imputer = KNNImputer(n_neighbors=5)
            df[numeric_columns] = imputer.fit_transform(df[numeric_columns])
            df[numeric_columns] = df[numeric_columns].round().astype(int)

            print("\n数值型字段缺失值处理完成")

        string_columns = ['天气情况', '风向', '风力等级']
        for col in string_columns:
            if df[col].isnull().any():
                mode_value = df[col].mode()[0]
                df[col].fillna(mode_value, inplace=True)
                print(f"\n字符型字段 {col} 缺失值处理情况:")
                print(f"众数: {mode_value}")

        # 处理异常值
        for col in numeric_columns:
            Q1 = df[col].quantile(0.25)
            Q3 = df[col].quantile(0.75)
            IQR = Q3 - Q1
            lower_bound = Q1 - 1.5 * IQR
            upper_bound = Q3 + 1.5 * IQR

            median_value = df[col].median().round().astype(int)
            df.loc[(df[col] < lower_bound) | (df[col] > upper_bound), col] = median_value

            print(f"\n字段 {col} 异常值处理情况:")
            print(f"中位数: {median_value}, 下界: {lower_bound}, 上界: {upper_bound}")

        # 处理重复值
        print("\n重复值情况:")
        print(df.duplicated().sum())
        df.drop_duplicates(inplace=True)

        print("\n清洗后的数据:")
        print(df.head())
        print("\n数据基本信息:")
        df.info()
        print("\n缺失值情况:")
        print(df.isnull().sum())

        insert_sql = """
        INSERT INTO cleaned_weatherdata (
            城市, 时间, 温度, 体感温度, 天气情况,
            风向角度, 风向, 风力等级, 风速, 湿度, 降水量, 大气压强, 能见度
        ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
        """
        for _, row in df.iterrows():
            data = (
                row['城市'],
                row['时间'],
                row['温度'],
                row['体感温度'],
                row['天气情况'],
                row['风向角度'],
                row['风向'],
                row['风力等级'],
                row['风速'],
                row['湿度'],
                row['降水量'],
                row['大气压强'],
                row['能见度']
            )
            with connection.cursor() as cursor:
                cursor.execute(insert_sql, data)
            connection.commit()

    except Exception as e:
        print(f"数据清洗时出错: {e}")
    finally:
        connection.close()

def create_weatherdata_table(db_config):
    """创建 weatherdata 表"""
    try:
        connection = pymysql.connect(**db_config)
        create_table_sql = """
        CREATE TABLE IF NOT EXISTS weatherdata (
            城市 VARCHAR(255) NOT NULL,
            时间 DATETIME,
            温度 INT,
            体感温度 INT,
            天气情况 VARCHAR(255),
            风向角度 INT,
            风向 VARCHAR(255),
            风力等级 VARCHAR(255),
            风速 INT,
            湿度 INT,
            降水量 INT,
            大气压强 INT,
            能见度 INT,
            记录时间 TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        """
        with connection.cursor() as cursor:
            cursor.execute(create_table_sql)
        connection.commit()
        print("weatherdata 表已创建")
    except Exception as e:
        print(f"创建 weatherdata 表时出错: {e}")
    finally:
        connection.close()

def create_cleaned_weatherdata_table(db_config):
    """创建 cleaned_weatherdata 表"""
    try:
        connection = pymysql.connect(**db_config)
        create_table_sql = """
        CREATE TABLE IF NOT EXISTS cleaned_weatherdata (
            id INT AUTO_INCREMENT PRIMARY KEY,
            城市 VARCHAR(255) NOT NULL,
            时间 DATETIME,
            温度 INT,
            体感温度 INT,
            天气情况 VARCHAR(255),
            风向角度 INT,
            风向 VARCHAR(255),
            风力等级 VARCHAR(255),
            风速 INT,
            湿度 INT,
            降水量 INT,
            大气压强 INT,
            能见度 INT,
            记录时间 TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        """
        with connection.cursor() as cursor:
            cursor.execute(create_table_sql)
        connection.commit()
        print("cleaned_weatherdata 表已创建")
    except Exception as e:
        print(f"创建 cleaned_weatherdata 表时出错: {e}")
    finally:
        connection.close()

def truncate_table(table_name, db_config):
    """清空表数据"""
    try:
        connection = pymysql.connect(**db_config)
        truncate_sql = f"TRUNCATE TABLE {table_name}"
        with connection.cursor() as cursor:
            cursor.execute(truncate_sql)
        connection.commit()
        print(f"表 {table_name} 的数据已清空")
    except Exception as e:
        print(f"清空表 {table_name} 时出错: {e}")
    finally:
        connection.close()

# 主程序
if __name__ == "__main__":
    # 数据库连接配置
    db_config = {
        "host": "localhost",
        "user": "root",
        "password": "1234567",
        "database": "tianqi",
        "charset": "utf8mb4"
    }

    # 读取城市列表
    with open("china.json", "r", encoding="utf-8") as file:
        china_city_codes = json.load(file)

    # API密钥
    api_key = "7f2691d4327e48439de7d3b86437a625"

    # 调用天气数据处理流程
    weather_data_pipeline(china_city_codes, api_key, db_config)
    time.sleep(1)